package dao

import (
	"context"

	"bmember/internal/model"
)

// ScheduleDao 定时任务 dao
type ScheduleDao interface {
	// CreateSchedule 创建定时任务
	CreateSchedule(ctx context.Context, sch *model.Schedules) (int64, error)
	// UpdateScheduleById 更新定时任务
	UpdateScheduleById(ctx context.Context, sch *model.Schedules) error
	// LockWaitingSchedulesByType 获取并锁定待执行定时任务
	LockWaitingSchedulesByType(ctx context.Context, schType int8, limit int) ([]*model.Schedules, error)
	// UpdateSchedulesStatusByIds 更新定时任务状态
	UpdateSchedulesStatusByIds(ctx context.Context, schIds []int64, status int8) error
	// LoadWaitingSchedules 按 type 获取定时任务，并更新状态
	LoadWaitingSchedules(ctx context.Context, schType int8, limit int) ([]*model.Schedules, error)
}

type schDaoImpl struct {
	*DaoBase
	name string
}

var schDao ScheduleDao

// InitScheduleDao 初始化  dao 层
func InitScheduleDao() {
	schDao = &schDaoImpl{
		DaoBase: daoBase,
		name:    "schedule_dao",
	}
}

// GetScheduleDao 获得定时任务 dao 层
func GetScheduleDao() ScheduleDao {
	return schDao
}

// CreateSchedule 创建定时任务
func (dao *schDaoImpl) CreateSchedule(ctx context.Context, sch *model.Schedules) (int64, error) {
	if sch == nil {
		return 0, nil
	}
	param := []interface{}{
		sch.SchType,
		sch.Retry,
		sch.Status,
		sch.Param,
		sch.ExecuteTime,
	}
	query := "INSERT INTO schedules (sch_type, retry, status, param, execute_time, created_at, updated_at) VALUES " +
		"(?,?,?,?,?,NOW(), NOW())"
	aff, err := dao.db.PgExec(ctx, "id", query, param...)
	if err != nil {
		return 0, err
	}
	lastId, _ := aff.GetLastId()
	return lastId, nil
}

// UpdateScheduleById 更新定时任务
func (dao *schDaoImpl) UpdateScheduleById(ctx context.Context, sch *model.Schedules) error {
	if sch == nil {
		return nil
	}
	param := []interface{}{
		sch.Retry,
		sch.Status,
		sch.ExecuteTime,
		sch.ErrMsg,
		sch.Id,
	}
	query := "UPDATE schedules SET retry=?, status=?, execute_time=?, err_msg=?, updated_at=NOW() WHERE id=?"
	_, err := dao.db.Update(ctx, query, param...)
	return err
}

// LockWaitingSchedulesByType 获取并锁定待执行定时任务
func (dao *schDaoImpl) LockWaitingSchedulesByType(ctx context.Context, schType int8, limit int) ([]*model.Schedules, error) {
	query := "SELECT id, sch_type, retry, status, param, execute_time FROM schedules WHERE status=1 AND execute_time<=NOW() " +
		" AND sch_type=? LIMIT ? FOR UPDATE"
	var schs []*model.Schedules
	err := dao.db.Query(ctx, &schs, query, schType, limit)
	return schs, err
}

// UpdateSchedulesStatusByIds 更新定时任务状态
func (dao *schDaoImpl) UpdateSchedulesStatusByIds(ctx context.Context, schIds []int64, status int8) error {
	if len(schIds) == 0 {
		return nil
	}
	query := "UPDATE schedules SET status=?, updated_at=NOW() WHERE id IN ?"
	_, err := dao.db.Update(ctx, query, status, schIds)
	return err
}

// LoadWaitingSchedules 按 type 获取定时任务，并更新状态
func (dao *schDaoImpl) LoadWaitingSchedules(ctx context.Context, schType int8, limit int) ([]*model.Schedules, error) {
	ret, err := dao.db.StartCapsule(ctx, true, func(ctx context.Context) (interface{}, error) {
		schs, err := dao.LockWaitingSchedulesByType(ctx, schType, limit)
		if err != nil {
			return nil, err
		}
		ids := make([]int64, len(schs))
		for idx, s := range schs {
			ids[idx] = s.Id
			s.Status = model.ScheduleDoing
		}
		err = dao.UpdateSchedulesStatusByIds(ctx, ids, model.ScheduleDoing)
		if err != nil {
			return nil, err
		}
		return schs, nil
	})
	if err != nil {
		return nil, err
	}
	return ret.([]*model.Schedules), nil
}
